package com.hins.mq.confirm;

import com.hins.mq.common.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;

/**
 * 消息确认异步模式
 */
public class Send2 {
    private final static String QUEUE_NAME = "confirm2_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道，这是完成大部分API的地方。
        Channel channel = connection.createChannel();

        // 声明（创建）队列，必须声明队列才能够发送消息，我们可以把消息发送到队列中。
        // 声明一个队列是幂等的 - 只有当它不存在时才会被创建
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 生产者调用confirmSelect 将channel设置为confirm模式
        channel.confirmSelect();

        //存储未确认的消息标识tag
        final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());

        channel.addConfirmListener(new ConfirmListener() {

            /*
             * 消息确认返回成功
             * l:如果是多条，这个就是最后一条消息的tag
             * b:是否多条
             * */
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                System.out.println("消息发送成功"+l+"是否多条"+b);
                if(b){
                    confirmSet.headSet(l+1).clear();
                }else{
                    confirmSet.remove(l);
                }
            }

            /*消息确认返回失败*/
            @Override
            public void handleNack(long l, boolean b) throws IOException {
                System.out.println("消息发送失败"+l+"是否多条"+b);
                if(b){
                    confirmSet.headSet(l+1).clear();
                }else{
                    confirmSet.remove(l);
                }
            }
        });


        for (int i = 0; i <10; i++) {
            String message = "hello ps"+i;
            long tag = channel.getNextPublishSeqNo();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes("utf-8"));
            System.out.println(tag);
            confirmSet.add(tag);
        }


        channel.close();
        connection.close();
    }
}